What if Test: 30 Mounths of Turtle Trade Test

Share

Big‑picture summary (Turtle Trading Style)

Time Span for Test: 1/1/2023 to 30/06/2025

Goal: Trade a Turtle‑style breakout on a top‑25, high‑liquidity universe using daily data.
Entry: Go long when the current price breaks above the rolling 30‑day highest close.

Exit (priority order):
Stop‑loss at −25% from average entry price.
Take‑profit at +30% from average entry price.
Exit if price falls below the rolling 15‑day lowest close.

Position sizing: Target 20% of portfolio value per position.

——— QuantConnect Code Generated by ChatGPT is below ———-

BackTest results from QuantConnect:
https://www.quantconnect.cloud/backtest/eebcd7650ae287029cf1d929f66de19e/?theme=chrome

Embed Code:
https://www.quantconnect.cloud/backtest/eebcd7650ae287029cf1d929f66de19e/?theme=chrome

In case the result links can be lost I copied the result front end.
As you can see 1.000.000 USD Becomes 2.718.001 USD in 30 months in test cycle.

from AlgorithmImports import *
from datetime import timedelta
from QuantConnect.Data.Consolidators import TradeBarConsolidator

class TurtleSP500(QCAlgorithm):
    def Initialize(self):
        # --- Backtest setup ---
        self.SetStartDate(2023, 1, 1)
        self.SetEndDate(2025, 6, 30)
        self.SetCash(1_000_000)

        # Warm up enough bars for Donchian windows
        self.SetWarmUp(60, Resolution.Daily)
        self.SetBrokerageModel(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)

        # --- Universe ---
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.CoarseSelectionFunction)

        # --- Strategy parameters ---
        self.entryLen = 30   # 30-day high for entry (close-based)
        self.exitLen  = 15    # 15-day low for exit (close-based)
        self.positionSize = 0.20  # 10% per trade

        # --- 1% TP/SL parameters ---
        self.takeProfitPct = 0.30   # +20% from avg entry
        self.stopLossPct   = 0.25   # -20% from avg entry

        # --- Storage ---
        self.symbolData = {}

        # --- Schedule daily trading ---
        self.Schedule.On(
            self.DateRules.EveryDay(),
            self.TimeRules.AfterMarketOpen("SPY", 30),
            self.ScanAndTrade
        )

    # === Universe selection: Top 25 by dollar volume ===
    def CoarseSelectionFunction(self, coarse):
        sorted_by_dollar_volume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True)
        top = sorted_by_dollar_volume[:25]
        return [c.Symbol for c in top]

    # === Handle securities added/removed from universe ===
    def OnSecuritiesChanged(self, changes):
        for sec in changes.AddedSecurities:
            symbol = sec.Symbol
            if symbol not in self.symbolData:
                self.symbolData[symbol] = SymbolData(self, symbol, self.entryLen, self.exitLen)

        for sec in changes.RemovedSecurities:
            symbol = sec.Symbol
            if symbol in self.symbolData:
                # Clean up consolidator and liquidate if needed
                self.symbolData[symbol].Dispose()
                self.Liquidate(symbol)
                del self.symbolData[symbol]

    # === Daily scan & trade logic ===
    def ScanAndTrade(self):
        if self.IsWarmingUp:
            return

        for symbol, data in list(self.symbolData.items()):
            if symbol not in self.Securities:
                continue

            security = self.Securities[symbol]
            if not data.IsReady or not security.IsTradable:
                continue

            price   = security.Price
            upper   = data.entryDonchianHigh()   # 20-day highest close (rolling)
            lower   = data.exitDonchianLow()     # 5-day lowest close (rolling)
            holding = self.Portfolio[symbol]
            invested = holding.Invested

            # --- Entry: price breaks above 20-day high ---
            if not invested and price > upper:
                self.SetHoldings(symbol, self.positionSize)
                self.Debug(f"BUY {symbol.Value} @ {price:.2f}")

            # --- Exits: 1% stop-loss, 1% take-profit, or Donchian 5-day low ---
            elif invested:
                avg_price = holding.AveragePrice
                if avg_price and avg_price > 0:
                    sl_level = avg_price * (1 - self.stopLossPct)
                    tp_level = avg_price * (1 + self.takeProfitPct)

                    # Precedence: stop-loss -> take-profit -> Donchian
                    if price <= sl_level:
                        self.Liquidate(symbol)
                        self.Debug(f"SELL {symbol.Value} @ {price:.2f} [StopLoss {self.stopLossPct*100:.0f}%]")
                        continue

                    if price >= tp_level:
                        self.Liquidate(symbol)
                        self.Debug(f"SELL {symbol.Value} @ {price:.2f} [TakeProfit {self.takeProfitPct*100:.0f}%]")
                        continue

                # Donchian exit (secondary)
                if price < lower:
                    self.Liquidate(symbol)
                    self.Debug(f"SELL {symbol.Value} @ {price:.2f} [Donchian-{self.exitLen}d]")


# === Helper class for Donchian channels (with daily rolling updates) ===
class SymbolData:
    def __init__(self, algo: QCAlgorithm, symbol: Symbol, entryLen: int, exitLen: int):
        self.algo = algo
        self.symbol = symbol
        self.entryLen = entryLen
        self.exitLen = exitLen

        # Close-based windows (match original definition)
        self.entryWindow = RollingWindow[float](entryLen)  # forighest close
        self.exitWindow  = RollingWindow[float](exitLen)   # for 5-day lowest       # 1) Seed with historical daily bars (DataFrame -> iterate itertuples)
        lookback = max(entryLen, exitLen) + 1
        history = algo.History(symbol, lookback, Resolution.Daily)
        if not history.empty:
            for bar in history.itertuples():
                # bar.close is a pandas namedtuple field
                self._add_close(float(bar.close))

        # 2) Add a daily consolidator so the windows roll forward each day
        self.consolidator = TradeBarConsolidator(timedelta(days=1))
        self.consolidator.DataConsolidated += self._on_daily_bar
        algo.SubscriptionManager.AddConsolidator(symbol, self.consolidator)

    def _on_daily_bar(self, sender, bar: TradeBar):
        self._add_close(float(bar.Close))

    def _add_close(self, close: float):
        self.entryWindow.Add(close)
        self.exitWindow.Add(close)

    @property
    def IsReady(self) -> bool:
        return (
            self.entryWindow.Count == self.entryWindow.Size and
            self.exitWindow.Count == self.exitWindow.Size
        )

    # --- Donchian methods (close-based) ---
    def entryDonchianHigh(self) -> float:
        return max(x for x in self.entryWindow)

    def exitDonchianLow(self) -> float:
        return min(x for x in self.exitWindow)

    def Dispose(self):
        # Clean up the consolidator when symbol leaves the universe
        if getattr(self, "consolidator", None) is not None:
            self.algo.SubscriptionManager.RemoveConsolidator(self.symbol, self.consolidator)
            self.consolidator = None

Share

Leave a Comment

This site uses Akismet to reduce spam. Learn how your comment data is processed.